package com.garlicts.framework.binlog.sync.canal;

import java.net.InetSocketAddress;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.google.common.collect.Lists;

@Component
public class CanalClient {
	
    private static final Logger logger = LoggerFactory.getLogger(CanalClient.class);
    
    private CanalConnector canalConnector;
    
    @Value("${canal.host}")
    String host;
    @Value("${canal.port}")
    String port;
    @Value("${canal.destination}")
    String destination;
    @Value("${canal.username}")
    String username;
    @Value("${canal.password}")
    String password;
    @Value("${canal.subscribe}")
    String subscribe;
    
    @Bean
    public CanalConnector getCanalConnector() {
    	
        canalConnector = CanalConnectors.newClusterConnector(
        		Lists.newArrayList(
        				new InetSocketAddress(host, Integer.valueOf(port))), destination, username, password);
        canalConnector.connect();
//        canalConnector.subscribe();
        // 回滚寻找上次中断的位置
//        canalConnector.rollback();
//        logger.info("canal客户端启动成功");
        return canalConnector;
    }

//    @Override
//    public void destroy() throws Exception {
//        if (canalConnector != null) {
//            canalConnector.disconnect();
//        }
//    }
    
}
